package com.atguigu.realtime.app.dwd;

import com.atguigu.realtime.app.BaseSQLApp;
import com.atguigu.realtime.common.Constant;
import com.atguigu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class Dwd_OrderInfo extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_OrderInfo().init(
                4002,
                2,
                "Dwd_OrderInfo"
        );
    }
    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv, "Dwd_OrderInfo");
//        tEnv.sqlQuery("select * from ods_db").execute().print();

        Table orderInfo = tEnv.sqlQuery("select " +
                " `data`['id'] id, " +
                " `data`['user_id'] user_id, " +
                " `data`['origin_amount'] origin_amount," +
                " `data`['coupon_reduce'] coupon_reduce," +
                " `data`['final_amount'] final_amount," +
                " `data`['order_status'] order_status," +
                " `data`['out_trade_no'] out_trade_no," +
                " `data`['trade_body'] trade_body," +
                " `data`['session_id'] session_id," +
                " `data`['province_id'] province_id," +
                " `data`['create_time'] create_time," +
                " `data`['expire_time'] expire_time," +
                " `data`['update_time'] update_time " +
                "from ods_db " +
                "where `database`='edu' " +
                "and " +
                "`table`='order_info' ");
        tEnv.createTemporaryView("order_info", orderInfo);

        tEnv.executeSql("create table dwd_order_info( " +
                "id string, " +
                "user_id string, " +
                "origin_amount string," +
                "coupon_reduce string," +
                "final_amount string," +
                "order_status string," +
                "out_trade_no string," +
                "trade_body string," +
                "session_id string," +
                "province_id string," +
                "create_time string," +
                "expire_time string," +
                "update_time string" +
                ")" + SQLUtil.getKafkaSink(Constant.TOPIC_DWD_ORDER_INFO));

        orderInfo.executeInsert("dwd_order_info");
    }
}
